package spark.core.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object rddSerializable {
  // Todo RDD 序列化

  // 1) 闭包检查
  // 从计算的角度, 算子以外的代码都是在Driver端执行, 算子里面的代码都是在Executor端执行
  // 那么在scala的函数式编程中，就会导致算子内经常会用到算子外的数据，这样就形成了闭包的效果，如果使用的算子外的数据无法序列化，就意味着无法传值给Executor端执行，就会发生错误
  // 所以需要在执行任务计算前，检测闭包内的对象是否可以进行序列化，这个操作我们称之为闭包检测。

  // 2) 序列化方法和属性
  // 从计算的角度, 算子以外的代码都是在Driver端执行, 算子里面的代码都是在Executor端执行

  def main(args: Array[String]): Unit = {
    // 1.创建SparkConf并设置App名称
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("rddSerializable")
    // 2. 创建SparkContext，该对象是提交Spark App的入口
    val sparkContext: SparkContext = new SparkContext(sparkConf)
    // 3. 创建一个RDD
    val RDD: RDD[String] = sparkContext.makeRDD(Array("hello world", "hello spark", "hive", "scala"))
    // 3.1创建一个Search对象
    val search: Search = new Search("h")
    // 3.2 函数传递，打印：ERROR Task not serializable
    search.getMatch1(RDD).collect().foreach(println)
    // 3.3 属性传递，打印：ERROR Task not serializable
    search.getMatch2(RDD).collect().foreach(println)
    //4.关闭连接
    sparkContext.stop()

  }

  private class Search(query: String) extends Serializable {
    private def isMatch(s: String): Boolean = {
      s.contains(query)
    }

    // 函数序列化案例
    def getMatch1(rdd: RDD[String]): RDD[String] = {
      //rdd.filter(this.isMatch)
      rdd.filter(isMatch)
    }

    // 属性序列化案例
    def getMatch2(rdd: RDD[String]): RDD[String] = {
      //rdd.filter(x => x.contains(this.query))
      rdd.filter(x => x.contains(query))
      //val q = query
      //rdd.filter(x => x.contains(q))
    }
  }
}
